// Copyright(C) 2023 InfiniFlow, Inc. All rights reserved.
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//     https://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.

export module infinity_core:physical_scan_base;

import :physical_operator;
import :physical_operator_type;
import :global_block_id;
import :base_table_ref;
import :load_meta;
import :operator_state;
import :query_context;
import :meta_info;

import internal_types;

namespace infinity {

class ResultCacheManager;
struct DataBlock;
struct BlockIndex;

export class PhysicalScanBase : public PhysicalOperator {
public:
    PhysicalScanBase(const u64 id,
                     const PhysicalOperatorType type,
                     std::unique_ptr<PhysicalOperator> left,
                     std::unique_ptr<PhysicalOperator> right,
                     const u64 table_index,
                     std::shared_ptr<BaseTableRef> base_table_ref,
                     std::shared_ptr<std::vector<LoadMeta>> load_metas,
                     const bool cache_result = false)
        : PhysicalOperator(type, std::move(left), std::move(right), id, std::move(load_metas), cache_result), table_index_(table_index),
          base_table_ref_(std::move(base_table_ref)) {}

    virtual std::vector<std::shared_ptr<std::vector<GlobalBlockID>>> PlanBlockEntries(i64 parallel_count) const;

    size_t TaskletCount() override;

    virtual BlockIndex *GetBlockIndex() const;

    void FillingTableRefs(std::unordered_map<size_t, std::shared_ptr<BaseTableRef>> &table_refs) override {
        table_refs.insert({base_table_ref_->table_index_, base_table_ref_});
    }

    u64 table_index() const { return table_index_; }

    [[nodiscard]] inline std::string TableAlias() const { return base_table_ref_->alias_; }

    [[nodiscard]] inline TableInfo *table_info() const { return base_table_ref_->table_info_.get(); }

protected:
    void SetOutput(const std::vector<char *> &raw_result_dists_list,
                   const std::vector<RowID *> &row_ids_list,
                   size_t result_size,
                   i64 result_n,
                   QueryContext *query_context,
                   OperatorState *operator_state) const;

    void
    AddCache(QueryContext *query_context, ResultCacheManager *cache_mgr, const std::vector<std::unique_ptr<DataBlock>> &output_data_blocks) const;

public:
    u64 table_index_ = 0;
    std::shared_ptr<BaseTableRef> base_table_ref_{};
};

} // namespace infinity
